Why did you add blocking-buffer here? Data Prepper API provides the Buffer interface.

I added this for testing S3Source using buffer.

For testing could this import be added as testImplementation project(':data-prepper-plugins:blocking-buffer')?

@sbayer55 's suggestion is the right approach for a testing dependency.

This will exit all of Data Prepper. That is a far too severe consequence of an error communicating with SQS. I think that we want to log the exception, then wait some time, then try again. It could be appropriate to perform an exponential backoff.

I recommend renaming this so that it has a different verb than "get." Perhaps "convert" or "transform?" Get makes it seem that you are using a getter.

I think this default should be 20 seconds (the current maximum).

These should be private static final (They are missing the static).

Your choice here: You could make these defaults actual Duration objects. Since they will be static, you won't need to re-create the Duration in each field. I also think it is cleaner since the code only deals with Duration.

e.g.

private static final Duration DEFAULT_VISIBILITY_TIMEOUT = Duration.ofSeconds(20);

I like this suggestion.

Why is this public and not private?

I think we will want to have a List<Thread> as a field. You can keep all the threads there. You will need to stop them when the stop() method for S3Source is called.

In general you should avoid logging just an error message from some external source. It is better to have:

LOG.error("Error reading from SQS: {}", e.awsErrorDetails().errorMessage());

Relatedly - is awsErrorDetails() guaranteed to be non-null?

I'm not sure if awsErrorDetails() will be non-null. I'll check this one

We should log this, not call e.printStackTrace().

For testing could this import be added as testImplementation project(':data-prepper-plugins:blocking-buffer')?

In place of utils could BackoffUtils class have a more descriptive name? Consider something like RetryState or BackoffManager. Without looking at the class file I expected BackoffUtils to contain some helper functions for parsing or transforming data.

It there a benefit to a random wait opposed to a fixed time or exponential growth time?

Yeah, it's to break the synchronization across the clients thereby avoiding collisions

Can the fully qualified name be shortened from software.amazon.awssdk.services.s3.S3Client to S3Client? It looks like you already have the import.

Why are we opting to build our own exponential back off strategy? Is there a reason we cannot use the built-in AWS SDK retry functionality?

We do need a different mechanism in place for the entire thread. As the SQS processing runs in its own thread, it must not end until the the source is shutdown.

I think we should use both:

• AWS SDK retry to help with transient errors
• Catch exceptions in the thread, suppress, and wait. This goes beyond AWS SDK errors. Say, some message in the SQS queue is not actually JSON. We don't want the thread to fail for a single bad message.

That being said, perhaps this will not need any sophisticated back-off. Perhaps a straightforward Thread.sleep() is enough.

This function is getting very large in this partial state. I would encourage you to break it apart to improve maintainability.

This will also make it easier to unit test and can remove variable reassignment/mutation in the code below.

@sbayer55 's suggestion is the right approach for a testing dependency.

We do need a different mechanism in place for the entire thread. As the SQS processing runs in its own thread, it must not end until the the source is shutdown.

I think we should use both:

• AWS SDK retry to help with transient errors
• Catch exceptions in the thread, suppress, and wait. This goes beyond AWS SDK errors. Say, some message in the SQS queue is not actually JSON. We don't want the thread to fail for a single bad message.

That being said, perhaps this will not need any sophisticated back-off. Perhaps a straightforward Thread.sleep() is enough.

Data Prepper core is calling the stop() method. Thus, this current code is going to interrupt the Data Prepper core thread.

Also, rather than use interrupt(), I'd like to have a running/shutdown boolean flag in the worker thread. You then can use while(running) { ... } instead of while(true) { ... }.

Is there a reason you can't use a mock here?

There may be value in creating some integration tests in other PRs which use the blocking buffer. But for a unit test, we should use mocks if at all possible.

I think we may want to make this part a separate class which is injected. One reason is that the SQS message may actually be in a different format in some cases. If the SQS queue is connected to SNS, then this becomes a string rather than JSON. So we might need to make this configurable by pipeline authors. I'm fine to keep it here for now if that helps merge it in though.

I think we can move this to a separate class once we decide on how pipeline authors can configure this.

I'd like to create an S3EventFilter structure. There are couple motivations I have in this:

1. We are also going to add an optional user-configuration to support filtering by bucketName or accountId.
2. If we add SNS as an option in future iterations, this filter should be the same code.

Here is a possible design:

public interface S3EventFilter {
  Optional<S3EventNotification.S3EventNotificationRecord> filter(S3EventNotification.S3EventNotificationRecord notification)
}

class ObjectCreatedFilter implements S3EventFilter {
  Optional<S3EventNotification.S3EventNotificationRecord> filter(S3EventNotification.S3EventNotificationRecord entry) {
    if (entry.getValue().getEventName().startsWith("ObjectCreated"))
      return Optional.of(entry);
    else
      return Optional.empty();
  }
}

We can build more complicated filters for the bucketName and accountId in another PR.

You might get a merge conflict on this file since I also added this into main. Please use the one I provided as it has toString() and is unit tested.

The buffer is never used the by the S3Service or the SqsService. Who is going to be responsible for putting the item in the buffer (S3Service or SqsService)?

The S3ObjectWorker handles this:

Also, @asifsmohammed and I are coordinating on the work here. He created this PR before the S3ObjectWorker was present in main, so some portions like this are currently unused in his branch and work.

I'll update the logic with buffer here

The run method is not tested. We are missing on out some core test coverage should add test cases.

This will also make it easier to unit test and can remove variable reassignment/mutation in the code below.

These private variables should be final to prevent re-assignment

I have a few concerns over this method and subsequent workflow. I have some ideas for improvements but I want to understand a few things first.

What are we adding an S3Object to? I find this method name confusing and was really confused by the usage in the run() method.

Why are the parameters and return object the same object (S3ObjectReference)? Are we extending S3ObjectReference and expecting this class to mutate the variable with a messageId and Receipt? (I want to see if we can avoid mutation and nullable variables)

How is the receipt used by the calling function? Do we need it?

Why are the parameters and return object the same object (S3ObjectReference)? Are we extending S3ObjectReference and expecting this class to mutate the variable with a messageId and Receipt? (I want to see if we can avoid mutation and nullable variables)

@asifsmohammed and I discussed a couple approaches here:

1. We can make this templated and add something like T getNotificationMetadata() which in this case returns SQS metadata. It could be extended for SNS metadata if that were ever added.
2. We could extend this class with a sub-class.

In either case, the SQS metadata would include the ReceiptHandle and MessageId.

I'm fine with either approach. But, the inheritance approach would have the benefit of completely abstracting this metadata from the S3 code.

My initial thought was to add ReceiptHandle and MessageId to S3ObjectReference which will be used to get S3 Object.
If S3 object is successfully processed we return the same S3ObjectReference to delete message from queue.

I'll create a subclass which contains ReceiptHandle and MessageId.